package comm.receivedata;

import com.google.gson.reflect.TypeToken;
import comm.SocketUtil.ServerSocketBase;
import comm.SocketUtil.ServerThread;
import comm.SocketUtil.SocketMapClass;
import comm.bean.GroupBean;
import exsun.jt808.data.base.MsgDirection;
import exsun.jt808.data.base.StatusResult;
import exsun.jt808.data.config.CommonMsgTopic;
import exsun.jt808.data.config.jt808.Jt808MsgIdTopicDic;
import exsun.jt808.data.config.jt808.Jt808ReceiveMsgTopic;
import exsun.jt808.data.distribute_data.DistributeEntity;
import exsun.jt808.data.enume.OrderSendStatus;
import exsun.jt808.data.kafka_data.KafkaJT808Data;
import exsun.jt808.data.util.GsonUtils;
import exsun.jt808.data.util.HostUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.io.BufferedWriter;
import java.io.OutputStreamWriter;
import java.lang.reflect.Type;
import java.net.Socket;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;

@Component
public class RawDataStorage {

    private static final Logger logger = LoggerFactory.getLogger(RawDataStorage.class);
    //    @Autowired
//    ServerSocketClass serverSocketClass;
    @Autowired
    GroupBean groupBean;


    Socket mapKey;
    Set<String> mapValue;



    //设备上行数据和平台下发原始数据都会在这里topic中
    @KafkaListener(id = "${handlecenter.kafka.consumer.RawDataMessageType.id}", topics = Jt808ReceiveMsgTopic.rawDtaResponse, containerFactory = "RawDataMessageTypeResponseFactory")
    private void channelWriteBatchRecordConsumerListen(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        try {
            //读取一行的数据
            for (ConsumerRecord<?, ?> record : records) {
                //获取的数据
                String msg = record.value().toString();
//                logger.info("接收上传数据:{}", msg);
                //获取的数据类型
                Type type = new TypeToken<KafkaJT808Data>() {
                }.getType();
                //设备上行数据   数据体
                KafkaJT808Data kafkaJT808Data = GsonUtils.getObjectFromJson(msg, type);
//                之前版本
//                if (ServerSocketBase.cilent != null && ServerThread.split.size() != 0 && ServerSocketBase.isConnend) {
//                    for (String devNo : ServerThread.split) {
//                        if (Objects.equals(kafkaJT808Data.getDevNo(), devNo)) {
//                            BufferedWriter bufferedWriter = new BufferedWriter(new OutputStreamWriter(ServerSocketBase.cilent.getOutputStream()));
//                            logger.info("向客户端发送数据：" + kafkaJT808Data.getJsonFromObject() + ",发送地址：" + ServerSocketBase.cilent.getInetAddress() + ":" + ServerSocketBase.cilent.getPort() + "\n");
//                            bufferedWriter.write(kafkaJT808Data.getJsonFromObject() + "\n");
//                            bufferedWriter.flush();
//                        }
//                    }
//                }

                if (SocketMapClass.SocketMap != null && SocketMapClass.SocketMap.size() != 0) {
                    for (Map.Entry<Socket, Set<String>> entry : SocketMapClass.SocketMap.entrySet()) {
                          mapKey = entry.getKey();
                          mapValue = entry.getValue();
                        //判断map里面对应的客户端和消息ID是否都都有，并且返回套接字连接的写入部分部分不是关闭状态，就按照对应的连接和设备转发数据
                        if (mapKey != null && mapValue != null && ServerThread.split.size() != 0 && !mapKey.isOutputShutdown()) {
                            for (String devNo : mapValue) {
                                if (Objects.equals(kafkaJT808Data.getDevNo(), devNo)) {
                                    BufferedWriter bufferedWriter = new BufferedWriter(new OutputStreamWriter(mapKey.getOutputStream()));
                                    logger.info("向客户端发送数据：" + kafkaJT808Data.getJsonFromObject() + ",数据发送客户端地址：" + mapKey.getInetAddress() + ":" + mapKey.getPort() + "\n");
                                    bufferedWriter.write(kafkaJT808Data.getJsonFromObject() + "\n");
                                    bufferedWriter.flush();
                                }
                            }
                        } else {
                            assert mapKey != null;
                            logger.info("Socket的IP和端口：" + mapKey.getInetAddress() + ":" + mapKey.getPort() + ",Socket的写入状态：" + mapKey.isOutputShutdown());
                        }
                    }
                }
//                else
//                {
//                    logger.info("SocketMap为空null,或者SocketMap内容0条！！！");
//                }
            }
            ack.acknowledge();
        } catch (Exception ex) {
            logger.error("异常信息：" + ex.toString());
            if (SocketMapClass.SocketMap.containsKey(mapKey)) {
                SocketMapClass.SocketMap.remove(mapKey, mapValue);
                logger.info("移除客户端连接：" + mapKey.getInetAddress() + ":" + mapKey.getPort());
            }
        }
    }
}